package com.bigdata.paimon;

import lombok.extern.slf4j.Slf4j;
import org.apache.paimon.catalog.Catalog;
import org.apache.paimon.data.InternalRow;
import org.apache.paimon.reader.RecordReader;

/**
 * description
 *
 * @author Cyber
 * <p> Created By 2025/4/9
 * @version 1.0
 */
@Slf4j
public class PaimonTableOption {


    public static final String databaseName = "paimon";

    public static final String tableName = "paimon_table";

    public static void main(String[] args) {
        Catalog catalog = PaimonCatalog.getPaimonCatalog(PaimonCatalog.HIVE_CATALOG_NAME, PaimonCatalog.METASTORE_TYPE_HIVE);
        log.info("============= paimon catalog created success ================");

        if (!PaimonCatalog.createPaimonDatabase(catalog, databaseName)) {
            log.info("create paimon database {} failed", databaseName);
            return;
        }
        log.info("============= create paimon database {} success ================", databaseName);
        org.apache.paimon.schema.Schema.Builder schemaBuilder = org.apache.paimon.schema.Schema.newBuilder();
        // 启用动态分桶 【在batchWrite的时候可以不指定分桶数】
        schemaBuilder.option("bucket", "-1");
        schemaBuilder.primaryKey("id");
        schemaBuilder.partitionKeys("city");
        schemaBuilder.column("id", org.apache.paimon.types.DataTypes.INT());
        schemaBuilder.column("name", org.apache.paimon.types.DataTypes.STRING());
        schemaBuilder.column("age", org.apache.paimon.types.DataTypes.INT());
        schemaBuilder.column("city", org.apache.paimon.types.DataTypes.STRING());
        org.apache.paimon.schema.Schema schema = schemaBuilder.build();
        if (!PaimonCatalog.createPaimonTable(catalog, databaseName, tableName, schema)) {
            log.warn("create paimon table {} failed", tableName);
            PaimonCatalog.dropPaimonTable(catalog, databaseName, tableName);
        }
        log.info("============= create paimon table {} success ==============", tableName);

        java.util.List<org.apache.paimon.data.GenericRow> RowsDataList = new java.util.ArrayList<org.apache.paimon.data.GenericRow>() {{
            add(org.apache.paimon.data.GenericRow.of(1, org.apache.paimon.data.BinaryString.fromString("Alice"), 22, org.apache.paimon.data.BinaryString.fromString("北京市")));
            add(org.apache.paimon.data.GenericRow.of(2, org.apache.paimon.data.BinaryString.fromString("Bob"), 25, org.apache.paimon.data.BinaryString.fromString("上海市")));
            add(org.apache.paimon.data.GenericRow.of(3, org.apache.paimon.data.BinaryString.fromString("Emily"), 18, org.apache.paimon.data.BinaryString.fromString("天津市")));
            add(org.apache.paimon.data.GenericRow.of(4, org.apache.paimon.data.BinaryString.fromString("Cyber"), 23, org.apache.paimon.data.BinaryString.fromString("武汉市")));
            add(org.apache.paimon.data.GenericRow.of(5, org.apache.paimon.data.BinaryString.fromString("Tom"), 26, org.apache.paimon.data.BinaryString.fromString("重庆市")));
            add(org.apache.paimon.data.GenericRow.of(6, org.apache.paimon.data.BinaryString.fromString("Tony"), 26, org.apache.paimon.data.BinaryString.fromString("北京市")));
            add(org.apache.paimon.data.GenericRow.of(7, org.apache.paimon.data.BinaryString.fromString("Jame"), 24, org.apache.paimon.data.BinaryString.fromString("上海市")));
        }};
        if (!PaimonCatalog.batchWriteTableData(catalog, databaseName, tableName, RowsDataList)) {
            PaimonCatalog.dropPaimonTable(catalog, databaseName, tableName);
            PaimonCatalog.dropPaimonDatabase(catalog, databaseName);
            return;
        }
        log.info("============= batchWrite data to paimon table {} success ==============", tableName);
        RecordReader<InternalRow> datas = PaimonCatalog.batchReadTableData(catalog, databaseName, tableName);
        try {
            datas.forEachRemaining(
                    record -> {
                        log.info("row: {} {} {} {}", record.getInt(0), record.getString(1), record.getInt(2), record.getString(3));
                    }
            );
        } catch (java.io.IOException e) {
            throw new RuntimeException(e);
        }
        log.info("============= batchRead data from paimon table {} success ==============", tableName);
    }
}
